package com.study.mq;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.study.service.AlbumItemService;
import com.study.service.RankingService;
import com.study.service.TimeWindowService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.Instant;


/**
 * @author : wilbur.yuan
 * @date : 2018-11-16
 */
@Slf4j
@Component
public class OrderConfirmListener implements MessageListener {


    @Autowired
    private ObjectMapper objectMapper;

    @Autowired
    private RankingService rankingService;

    @Autowired
    private AlbumItemService albumItemService;

    @Autowired
    private TimeWindowService timeWindowService;


    @Override
    public void onMessage(Message message) {

        if (!timeWindowService.filter()) {
            log.info("not in the counting period , return");
            return;
        }
        MqMessage mqMessage = decrypt(message);

        Long msgId = mqMessage.getMsgId();

        String albumId = albumItemService.getAlbumIdByItemId(mqMessage.getItemId());
        if (StringUtils.isEmpty(albumId)) {
            log.info("msgId={}, itemId={} not in the counting list", msgId, mqMessage.getItemId());
            return;
        }

        log.info("fans rankings consume mes: " + new String(message.getBody(), StandardCharsets.UTF_8));

        String result = rankingService.addRanking(msgId,albumId, getRedisArgs(mqMessage));

        log.info("msgId={} ,redisScript result={}", msgId, result);
    }


    private MqMessage decrypt(Message message) throws MessageConversionException {
        try {
            String str = new String(message.getBody(),StandardCharsets.UTF_8);
            log.debug("msg : " + str);
            return objectMapper.readValue(str, MqMessage.class);
        } catch (IOException e) {
            throw new MessageConversionException("convert fail", e);
        }
    }



    private String[] getRedisArgs(MqMessage mqMessage) {
        String[] args = new String[4];
        args[0] = "1";
        int quantity = Math.toIntExact(mqMessage.getQuantity());
        args[1] = String.valueOf(quantity);
        args[2] = String.valueOf(mqMessage.getBuyerId());
        args[3] = String.valueOf(Instant.now().getEpochSecond());
        return args;
    }


}
